package com.zero.rocketmq.api.base.customer;

import com.zero.rocketmq.api.utils.RocketMQContents;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.PullResult;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageQueue;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;

/**
 * 消费者主动去Broker上拉取消息的拉模
 */
public class PullConsumer {

    private static final Map<MessageQueue, Long> OFFSE_TABLE = new HashMap<MessageQueue, Long>();

    public static void main(String[] args) throws MQClientException {

        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("PULLConsumer-Group");
        consumer.setNamesrvAddr("10.185.3.41:9876");

        consumer.start();

        //获取对应topic的所有MessageQueue
        Set<MessageQueue> messageQueues = consumer.fetchSubscribeMessageQueues(RocketMQContents.TOPIC);
        for (MessageQueue messageQueue : messageQueues){
            System.out.printf("Consume from the queue: %s%n", messageQueue);
            SINGLE_MQ:
            while (true){
                try{
                    PullResult pullResult = consumer.pullBlockIfNotFound(messageQueue, null, getMessageQueueOffset(messageQueue), 32);
                    System.out.printf("%s%n", pullResult);
                    //重置消息的offset
                    putMessageQueueOffset(messageQueue,pullResult.getNextBeginOffset());
                    switch (pullResult.getPullStatus()) {
                        case FOUND:
                            break;
                        case NO_MATCHED_MSG:
                            break;
                        case NO_NEW_MSG:
                            break SINGLE_MQ;
                        case OFFSET_ILLEGAL:
                            break;
                        default:
                            break;
                    }
                }catch (Exception e){
                    e.printStackTrace();
                }

            }
        }
        consumer.shutdown();
    }

    private static long getMessageQueueOffset(MessageQueue messageQueue){

        final Long offset = OFFSE_TABLE.get(messageQueue);
        if(offset != null){
            return offset;
        }
        return 0;
    }

    private static void putMessageQueueOffset(MessageQueue messageQueue,long offset){
        OFFSE_TABLE.put(messageQueue,offset);
    }
}
